package com.xdemo.processor;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;

import com.xdemo.handler.MessageActionRouter;
import com.xdemo.msg.BaseHttpMessage;
import com.xdemo.msg.IMessage;

public class ExecutorQueueProcessor<T> implements IProcessor<T>,Runnable {
	
	private final BlockingQueue<IMessage<T>> messages = new LinkedBlockingQueue<IMessage<T>>();
	
	private final ExecutorService executors;
	
	private final MessageActionRouter<T> router;
	
	private final Thread thread = new Thread(this);
	
	private boolean stoped;
	
	public ExecutorQueueProcessor(ExecutorService executors,MessageActionRouter<T> router) {
		this.executors = executors;
		this.router = router;
	}

	@Override
	public void start() {
		thread.start();
	}
	
	@Override
	public void stop() {
		stoped = true;
	}

	@Override
	public void execute(IMessage<T> message) {
		if(!(message instanceof BaseHttpMessage)) {
			return;
		}
		messages.add(message);
	}

	@Override
	public void run() {
		while(!stoped) {
			try {
				processor(messages.take());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private void processor(final IMessage<T> message) {
		executors.execute(new Runnable() {
			@Override
			public void run() {
				router.handleMessage(message);
			}
		});
	}
	
	public boolean isFull() {
		return messages.size() > 1000;
	}
}
